package main

import (
	"gitee.com/Cookie_XiaoD/easykafka"
	"gitee.com/Cookie_XiaoD/easykafka/spec"
	"log"
	"strconv"
	"time"
)

var brokers = "127.0.0.1:9092"

func main() {
	producer, err := easykafka.NewProducer(
		brokers,
		easykafka.WithProducerErrorHandler(func(err *easykafka.AsyncProduceError) {
			log.Println("异步生产消息时发生错误：", err)
		}),
		easykafka.WithProducerAckMode(spec.WaitLeader))
	if err != nil {
		log.Fatalf(err.Error())
	}
	defer func() {
		if err = producer.Close(); err != nil {
			log.Println("关闭生产者发生错误：", err)
		}
	}()
	seq := 1
	for {
		size, err := producer.SyncProduce("topic_example", "key", ExampleData{
			Seq:     strconv.Itoa(seq),
			Content: "消息" + time.Now().Format("2006-01-02 15:04:05.000"),
		})
		if err != nil {
			log.Println("发送消息错误：", err)
		} else {
			log.Println("发送成功，数据序号：", seq, "数据大小：", size)
			seq++
		}
		time.Sleep(1 * time.Second)
	}
}

type ExampleData struct {
	Content string `json:"content"`
	Seq     string `json:"seq"`
}
